package com.xormedia.httpservice.util;

import java.io.Closeable;
import java.util.HashSet;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
 * </br>
 * com.xormedia.aqua.federal.SynTaskExecutor
 * 
 * @author zhikang.peng
 * @date 2016年9月6日上午10:59:56
 * @since 1.0
 */
public class HandlerExecutor extends ThreadPoolExecutor {

  private static final Logger slf4jLogger = LoggerFactory.getLogger(HandlerExecutor.class);
  // 所有正在执行的任务集合
  private final HashSet<Runnable> tasks = new HashSet<Runnable>();
  private static final RejectedExecutionHandler defaultHandler = new WaitPolicy();

  public HandlerExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
      BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
    super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, defaultHandler);
  }

  @Override
  public void afterExecute(Runnable r, Throwable t) {
    tasks.remove(r);
  }

  @Override
  public void beforeExecute(Thread t, Runnable r) {
    tasks.add(r);
  }

  /**
   * 关闭线程池
   * 
   * @param pool
   * @author zhikang.peng
   * @date 2016年8月4日下午4:28:41
   * @since 1.0
   */
  public void shutdownAndAwaitTermination() {
    shutdown(); // Disable new tasks from being submitted
    try {
      // 通知调度器中所有未完成的任务结束任务，和清空等待队列中的任务
      for (Runnable r : tasks) {
        if (r instanceof Closeable) {
          ((Closeable) r).close();
        }
      }
      // 清空等待队列
      getQueue().clear();
      // Wait a while for existing tasks to terminate
      if (!awaitTermination(60, TimeUnit.SECONDS)) {
        shutdownNow(); // Cancel currently executing tasks
        // Wait a while for tasks to respond to being cancelled
        if (!awaitTermination(60, TimeUnit.SECONDS))
          slf4jLogger.error("Pool did not terminate");
      }
    } catch (Exception e) {
      // (Re-)Cancel if current thread also interrupted
      shutdownNow();
      // Preserve interrupt status
      Thread.currentThread().interrupt();
    }
  }

  /**
   * 正在执行任务的线程个数
   * 
   * @return
   * @author zhikang.peng
   * @date 2017年1月10日下午3:16:04
   * @since 1.0
   */
  public int taskSize() {
    return tasks.size();
  }

  static class WaitPolicy implements RejectedExecutionHandler {

    @Override
    public void rejectedExecution(Runnable task, ThreadPoolExecutor tpexecutor) {
      try {
        // 阻塞
        tpexecutor.getQueue().put(task);
      } catch (InterruptedException e) {
        slf4jLogger.error("", e);
      }
    }

  }

}
